package com.codejiwei.flink.cep;

import com.codejiwei.flink.entity.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @author jiwei
 * @description
 * @date 2023/5/26 11:18
 */
public class Flink_CEP_PatternAPI_FixTimes {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> stream = env.readTextFile("src/main/resources/data/sensor-data.log")
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String line, Collector<WaterSensor> collector) throws Exception {
                        String[] fields = line.split(",");
                        collector.collect(new WaterSensor(fields[0], Long.parseLong(fields[1]) * 1000L, Integer.valueOf(fields[2])));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor waterSensor, long l) {
                                return waterSensor.getTs();
                            }
                        })
                );

        //1. 定义模式
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                .<WaterSensor>begin("start")
                .where(new SimpleCondition<WaterSensor>() {
                    @Override
                    public boolean filter(WaterSensor waterSensor) throws Exception {
                        return waterSensor.getId().equals("sensor_1");
                    }
                });
        //2. 使用量词 出现3次
//        Pattern<WaterSensor, WaterSensor> times = pattern.times(3);
        //2. 范围内的次数 出现2-4次
//        Pattern<WaterSensor, WaterSensor> times = pattern.times(2, 4);
        //2. 一次或多次
//        Pattern<WaterSensor, WaterSensor> times = pattern.oneOrMore();
        //2. 多次及多次以上
        Pattern<WaterSensor, WaterSensor> times = pattern.timesOrMore(4);

        CEP.pattern(stream, times)
                .select(new PatternSelectFunction<WaterSensor, String>() {
                    @Override
                    public String select(Map<String, List<WaterSensor>> map) throws Exception {
//                        System.out.println("连续出现次数：" + );
                        return map.toString();
                    }
                })
                .print();


        env.execute();
    }
}
